package com.xxxx;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Hello09FlinkReduceFunction {
    public static void main(String[] args) throws Exception {
        //1.初始化环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        //2.读取数据源
        DataStreamSource<String> source = environment.socketTextStream("192.168.88.101", 18880);
        source.map(word -> {
            System.err.println(word + "--" + System.currentTimeMillis());
            return Tuple2.of(word, 1);
        }).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(0).countWindow(10).reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> tuple201, Tuple2<String, Integer> tuple202) throws Exception {
                return Tuple2.of(tuple201.f0, tuple201.f0.length() + tuple202.f0.length());
            }
        }).print();

        //3.需要手动触发流式计算的执行效果
        environment.execute("Hello09FlinkReduceFunction" + System.currentTimeMillis());
    }
}